package com.atguigu.flink.wordcount;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import static org.apache.flink.api.common.typeinfo.Types.*;

//把希望使用的静态方法和静态属性直接导入到当前类中，也称为静态导入


/**
 * Created by Smexy on 2023/1/12
 *
 *
 *      泛型: Collector<T>
 *              T是泛型
 *
 *      类型： Collector
 *
 *
 */
public class Demo7_LamdaTuple2
{
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        DataStreamSource<String> ds = env.socketTextStream("hadoop103",8888);

        SingleOutputStreamOperator<Tuple2<String, Integer>> ds1 = ds
            .flatMap((String line, Collector<Tuple2<String, Integer>> out) -> {
            String[] words = line.split(" ");
            for (String word : words) {
                out.collect(Tuple2.of(word, 1));
            }
        })
            /*
                returns(Class<T> typeClass): 只能提示非泛型类型。例如WordCount
                    如果是Tuple2<T0, T1>，含有泛型，要使用
                         returns(TypeHint)
                            或
                         returns(TypeInformation<T> typeInfo)
                            Types: flink提供的工具类，返回常见的类型的 TypeInformation 实例
             */
            //.returns(new TypeHint<Tuple2<String, Integer>>(){});
            .returns(TUPLE(STRING,INT));


        ds1.keyBy(t -> t.f0)
          .sum(1)
          .print();


        env.execute();


    }
}
